# @Version        : 1.0
# @Update Time    : 2024/9/8 22:59
# @File           : auth.py
# @IDE            : PyCharm
# @Desc           : 认证依赖
from typing import Optional, Literal, Callable
from fastapi import Depends, Request
from fastapi.security import OAuth2PasswordBearer

from src.utils.auth.auth_permission import SecurityUtils
from src.common.typing.common_typing import DatabaseSession
from src.core.exception import BaseAppException
from src.db.db_redis import AsyncRedisTool, get_redis_dependency
from src.schemas.user_schema import UserInfo

__all__ = ["OpenAuth", "AllUserAuth", "PermissionAuth", "RoleAuth"]

OAUTH2_SCHEMA: OAuth2PasswordBearer | Callable[[], str] = OAuth2PasswordBearer(
    tokenUrl="/auth/api/login", auto_error=False
)


class OpenAuth(SecurityUtils):
    async def __call__(
        self,
        request: Request,
        db: DatabaseSession,
        token: str = Depends(OAUTH2_SCHEMA),
        redis: AsyncRedisTool = Depends(get_redis_dependency),
    ) -> UserInfo | None:
        """
        开放认证依赖
        :param request:
        :param token:
        :param db:
        :return:
        """
        if token is None:
            return None
        # 测试校验用户信息，成功返回用户信息，失败返回数据库链接
        try:
            user = await self.get_user_info(
                request=request, db=db, token=token, redis=redis, is_perm=False
            )
            return user
        except BaseAppException:
            return None


class AllUserAuth(SecurityUtils):
    async def __call__(
        self,
        request: Request,
        db: DatabaseSession,
        token: str = Depends(OAUTH2_SCHEMA),
        redis: AsyncRedisTool = Depends(get_redis_dependency),
    ) -> UserInfo:
        """
        登录可访问
        :param request:
        :param token:
        :param db:
        :return:
        """
        user = await self.get_user_info(
            request=request, db=db, token=token, redis=redis, is_perm=False
        )
        return user


class PermissionAuth(SecurityUtils):
    def __init__(
        self,
        permissions: Optional[list[str]] = None,
        logical: Literal["and", "or"] = "and",
    ):
        """
        权限校验
        :param permissions: 校验权限
        :param logical: 校验逻辑
        """
        if permissions is None:
            self.permissions = None
        else:
            self.permissions = set(permissions)
        self.logical = logical

    async def __call__(
        self,
        request: Request,
        db: DatabaseSession,
        token: str = Depends(OAUTH2_SCHEMA),
        redis: AsyncRedisTool = Depends(get_redis_dependency),
    ) -> UserInfo:
        """

        :param request:
        :param token: token
        :param db: 获取一个具有事务的数据库连接
        :return:
        """
        user = await self.get_user_info(request, db, redis, token, is_perm=True)
        if self.permissions is not None and self.ALL_PERMISSION not in user.permissions:
            if self.logical == "and":
                await self.check_permi_and(self.permissions, set(user.permissions))
            else:
                await self.check_permi_or(self.permissions, set(user.permissions))
        return user


class RoleAuth(SecurityUtils):
    def __init__(
        self, roles: Optional[list[str]] = None, logical: Literal["and", "or"] = "and"
    ):
        """
        角色校验
        :param roles: 校验角色
        :param logical: 校验逻辑
        """
        if roles is None:
            self.roles = None
        else:
            self.roles = set(roles)
        self.logical = logical

    async def __call__(
        self,
        request: Request,
        db: DatabaseSession,
        token: str = Depends(OAUTH2_SCHEMA),
        redis: AsyncRedisTool = Depends(get_redis_dependency),
    ):
        """
        角色校验
        :param request:
        :param token:
        :param db:
        :return:
        """
        user = await self.get_user_info(request, db, redis, token)
        role_keys = set([role.role_key for role in user.roles])
        if self.roles is not None and self.SUPER_ADMIN not in role_keys:
            if self.logical == "and":
                await self.check_role_and(self.roles, role_keys)
            else:
                await self.check_role_or(self.roles, role_keys)
        return user
